package cn.com.dhcc.turbo.service.registry.sync;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import cn.com.dhcc.app.core.CM;
import cn.com.dhcc.app.core.exception.DaoException;
import cn.com.dhcc.app.core.exception.NiceException;
import cn.com.dhcc.app.core.exception.ServiceException;
import cn.com.dhcc.app.core.handler.cachehandler.CacheHandler.CacheOperType;
import cn.com.dhcc.app.core.handler.dbhandler.SingleTransationCircleWithOutResult;
import cn.com.dhcc.app.core.service.Log;
import cn.com.dhcc.app.core.service.Log.LOGTYPE;
import cn.com.dhcc.app.core.tag.CommonStaticCode.SUCCESSFAIL;
import cn.com.dhcc.app.core.tag.CommonStaticCode.YESNO;
import cn.com.dhcc.app.core.util.CollectionUtil;
import cn.com.dhcc.app.core.util.JsonUtil;
import cn.com.dhcc.app.core.util.UUIDGenerator;
import cn.com.dhcc.app.pub.core.consts.AppIocBean;
import cn.com.dhcc.app.pub.core.consts.NodeInfoParam;
import cn.com.dhcc.app.pub.core.service.TurboCacheService;
import cn.com.dhcc.app.pub.core.tag.AppTagEl;
import cn.com.dhcc.turbo.entity.exch.ExchRel;
import cn.com.dhcc.turbo.entity.exch.ExchRelPipeline;
import cn.com.dhcc.turbo.entity.exch.GapOwnerConf;
import cn.com.dhcc.turbo.entity.exch.NodeLayout;
import cn.com.dhcc.turbo.entity.monitor.QuotaValueOnTime;
import cn.com.dhcc.turbo.entity.registry.AdapterBase;
import cn.com.dhcc.turbo.entity.registry.AdapterInfo;
import cn.com.dhcc.turbo.entity.registry.DataCategory;
import cn.com.dhcc.turbo.entity.registry.DomainInfo;
import cn.com.dhcc.turbo.entity.registry.NodeInfo;
import cn.com.dhcc.turbo.entity.registry.NodePipeline;
import cn.com.dhcc.turbo.entity.registry.SynchronizeInfo;
import cn.com.dhcc.turbo.entity.registry.SynchronizeInfo.SyncOperType;
import cn.com.dhcc.turbo.entity.registry.SynchronizeInfo.SyncType;
import cn.com.dhcc.turbo.service.exch.ExchRelService;
import cn.com.dhcc.turbo.service.exch.cmdata.GapOwnerConfService;
import cn.com.dhcc.turbo.service.monitor.MonitorService;
import cn.com.dhcc.turbo.service.queue.tlq.TlqMoveService;
import cn.com.dhcc.turbo.service.registry.AdapterBaseService;
import cn.com.dhcc.turbo.service.registry.AdapterInfoService;
import cn.com.dhcc.turbo.service.registry.DataCategoryService;
import cn.com.dhcc.turbo.service.registry.DomainInfoService;
import cn.com.dhcc.turbo.service.registry.LocalConfCallBack;
import cn.com.dhcc.turbo.service.registry.NodeInfoService;
import cn.com.dhcc.turbo.service.registry.NodePipelineService;
import cn.com.dhcc.turbo.service.registry.PipelineRegionService;
import cn.com.dhcc.turbo.service.registry.SzyFeatureMgr;
import cn.com.dhcc.turbo.service.registry.vo.GapOwnerConfigSyncVo;
import cn.com.dhcc.turbo.service.registry.vo.NodeRegistSyncData;


@Service
public class SynchronizeHandler extends  UnicastRemoteObject implements ISynchronizeHandler{

	protected SynchronizeHandler() throws RemoteException {
		super();
	}
	private static final long serialVersionUID = 1L;
	@Autowired
	DomainInfoService domainInfoService;
	@Autowired
	NodeInfoService nodeInfoService;
	@Autowired
	PipelineRegionService pipelineRegionService;
	@Autowired
	NodePipelineService nodePipelineService;
	@Autowired
	ExchRelService exchRelService;
	@Autowired
	MonitorService monitorService;
	@Autowired
	SyncCenter syncCenter;
	@Autowired
	SzyFeatureMgr nodeRegistryPermissionMgr;
	@Autowired
	LocalConfCallBack localConfCallBack;
	@Autowired
	GapOwnerConfService gapOwnerConfService;
	@Autowired
	private AdapterInfoService adapterInfoService;
	
	@Autowired
	private AdapterBaseService adapterBaseService;
	@Autowired
	private DataCategoryService categoryService;
	
	@Autowired
	@Qualifier(AppIocBean.TURBO_CACHE_SERVICE)
	private TurboCacheService turboCacheService;
	
	private Logger log = Log.getLogger(LOGTYPE.SYNC);
	
	@Autowired
	private TlqMoveService tlqMoveService;
	@SuppressWarnings("unchecked")
	public SyncCallBack receiveData(Object receivedData)throws RemoteException{
		SyncCallBack syncCallBack = null;
		if(receivedData instanceof SynchronizeInfo){
			SynchronizeInfo synchronizeInfo = (SynchronizeInfo)receivedData;
			String dataType = synchronizeInfo.getDataType();
			String dataContent = synchronizeInfo.getDataContent();
			NodeInfo fromNode = synchronizeInfo.getFromNodeInfo();//从哪个节点同步过来的
			NodeInfo srcNode = synchronizeInfo.getSrcNodeInfo();//数据是谁维护的，有可能中转
			log.info("接收到同步信息，来自节点["+fromNode.getNodeName()+"],数据源节点["+srcNode+"],数据类型："+AppTagEl.getNameFromCode(SyncType.class, synchronizeInfo.getDataType())+",\r\n详细信息："+synchronizeInfo);
			if(SyncType.NODE_REGIST.getCode().equalsIgnoreCase(dataType)){//处理节点注册时的批量同步
				NodeRegistSyncData nodeRegistSyncData = JsonUtil.toObject(dataContent, NodeRegistSyncData.class);
				syncCallBack = receiveNodeRegistSyncData(fromNode,srcNode,nodeRegistSyncData);
			}else if(SyncType.NODE_INFO.getCode().equalsIgnoreCase(dataType)){//处理节点的增删改查同步
				NodeInfo srcNodeInfo = JsonUtil.toObject(dataContent, NodeInfo.class);
				syncCallBack = receiveNodeInfoSyncData(fromNode,srcNodeInfo);
			}else if(SyncType.DOMAIN_INFO.getCode().equalsIgnoreCase(dataType)){//处理域的增删改查同步
				DomainInfo srcDomainInfo = JsonUtil.toObject(dataContent, DomainInfo.class);
				syncCallBack = receiveDomainInfoSyncData(fromNode,srcNode,srcDomainInfo);
			}else if(SyncType.ADAPTER_BASE.getCode().equalsIgnoreCase(dataType)){//处理注册系统的增删改查同步
				AdapterBase adapterBase = JsonUtil.toObject(dataContent, AdapterBase.class);
				syncCallBack = receiveAdapterBaseSyncData(fromNode,srcNode,adapterBase);
			}else if(SyncType.DATACATEGORY.getCode().equalsIgnoreCase(dataType)){//处理注册系统的增删改查同步
				DataCategory dataCategory = JsonUtil.toObject(dataContent, DataCategory.class);
				syncCallBack = receiveDataCategorySyncData(fromNode,srcNode,dataCategory);
			}else if(SyncType.ADAPTER.getCode().equalsIgnoreCase(dataType)){//处理注册系统的增删改查同步
				AdapterInfo adapterInfo = JsonUtil.toObject(dataContent, AdapterInfo.class);
				syncCallBack = receiveAdapterInfoSyncData(fromNode,srcNode,adapterInfo);
			}else if(SyncType.PILE_LINE.getCode().equalsIgnoreCase(dataType)){//处理通道同步信息的同步
				NodePipeline srcPipeline = JsonUtil.toObject(dataContent, NodePipeline.class);
				syncCallBack = receivePipelineSyncData(fromNode,srcNode,srcPipeline);
			}else if(SyncType.PILE_LINE_LIST.getCode().equalsIgnoreCase(dataType)){//处理多个通道同步信息的同步
				List<NodePipeline> srcPipelines = JsonUtil.toObject(dataContent, List.class);
				syncCallBack = receivePipelineListSyncData(fromNode,srcNode,srcPipelines);
			}else if(SyncType.MONITOR.getCode().equalsIgnoreCase(dataType)){//处理管理范围内监控信息的同步
				QuotaValueOnTime quotaValueOnTime =JsonUtil.toObject(dataContent, QuotaValueOnTime.class);
				syncCallBack = receiveMonitorSyncData(srcNode,quotaValueOnTime);
			}else if(SyncType.CREATE_REL.getCode().equalsIgnoreCase(dataType)){//处理建立交换关系的同步
				ExchRel exchRel = JsonUtil.toObject(dataContent, ExchRel.class);
				syncCallBack = receiveCreateRelSyncData(exchRel);
			}else if(SyncType.DELEE_REL.getCode().equalsIgnoreCase(dataType)){//处理删除建立交换关系的同步
				ExchRel exchRel = JsonUtil.toObject(dataContent, ExchRel.class);
				syncCallBack = receiveDeleteRelSyncData(exchRel);
			}else if(SyncType.TEST_CONNECT.getCode().equalsIgnoreCase(dataType)){//连接测试
				syncCallBack = new SyncCallBack();
				syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
			}else if(SyncType.FLEX_LAYOUT.getCode().equalsIgnoreCase(dataType)){//flex 坐标信息
				NodeLayout nodeLayout = JsonUtil.toObject(dataContent, NodeLayout.class);
				syncCallBack = exchRelService.mergeNodeLayout(nodeLayout);
			}else if(SyncType.GAP_CONFIG.getCode().equalsIgnoreCase(dataType)){//处理网闸权限配置信息
				GapOwnerConfigSyncVo configSyncVo = JsonUtil.toObject(dataContent, GapOwnerConfigSyncVo.class);
				syncCallBack = receiveGapCongInfo(configSyncVo);
			}else if(SyncType.REL_PIPE_LINE.getCode().equalsIgnoreCase(dataType)){//需要获取通道信息
				ExchRelPipeline exchRelPipeline = JsonUtil.toObject(dataContent, ExchRelPipeline.class);
				syncCallBack = receiveExchRelPipeline(exchRelPipeline);
			}
			else{
				log.warn("收到不支持的同步信息，"+receivedData);
			}
			if(syncCallBack !=null && syncCallBack.getResult()==SUCCESSFAIL.SUCCESS){
				log.info("接收同步信息成功，来自节点["+fromNode.getNodeName()+"],数据源节点["+srcNode+"],数据类型："+AppTagEl.getNameFromCode(SyncType.class, synchronizeInfo.getDataType()));
			}else{
				log.error("接收同步信息失败，来自节点["+fromNode.getNodeName()+"],数据源节点["+srcNode+"],数据类型："+AppTagEl.getNameFromCode(SyncType.class, synchronizeInfo.getDataType()));
			}
		}else{
			log.warn("收到不支持的socket信息，"+receivedData);
			syncCallBack = new SyncCallBack();
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
		}
		
		return syncCallBack;
	}
	
	
	
	
	
	/**
	 * 接收数据分类信息
	 * @param fromNode
	 * @param srcNode
	 * @param dataCategory
	 * @return
	 */
	private SyncCallBack receiveExchRelPipeline(ExchRelPipeline exchRelPipeline){
		SyncCallBack syncCallBack = new SyncCallBack();
		try {
			NodePipeline line=null;
			if (exchRelPipeline.getNeed()!=null) {
				if (exchRelPipeline.getNeed().getId().equals(exchRelPipeline.getFromNode().getId())) {//如果需要的是fromNode的通道信息
					line=this.getPipleLineInfo(exchRelPipeline.getToNode().getNodeCode());//以toNode为参数查询
				}else{
					line=this.getPipleLineInfo(exchRelPipeline.getFromNode().getNodeCode());//以fromNode为参数查询
					this.updatePipeline(exchRelPipeline.getFromPipeline());//更新
				}
			}else{//仅仅更新就行
				this.updatePipeline(exchRelPipeline.getToPipeline());
			}
			syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
			syncCallBack.setNodePipeline(line);
		} catch (Exception e) {
			syncCallBack.setMsg("建立交换关系获取通道信息"+e.getMessage());
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
		}
		return syncCallBack;
	}
	
	
	
	
	
	/**
	 * 接收数据分类信息
	 * @param fromNode
	 * @param srcNode
	 * @param dataCategory
	 * @return
	 */
	private SyncCallBack receiveDataCategorySyncData(NodeInfo fromNodeInfo,
			NodeInfo srcNodeInfo, DataCategory dataCategory) {
		SyncCallBack syncCallBack = new SyncCallBack();
		try {
			categoryService.insertOrUpdate(dataCategory);
			syncToAllNodes(SyncType.DATACATEGORY,isFromParent(fromNodeInfo),fromNodeInfo,srcNodeInfo,dataCategory,SyncOperType.UPDATE,true);
			syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
		} catch (Exception e) {
			syncCallBack.setMsg("目的节点接收同步的数据分类信息失败,"+e.getMessage());
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
		}
		return syncCallBack;
	}
	
	/**
	 * 接收适配器基本信息同步信息
	 * @param fromNode
	 * @param srcNode
	 * @param adapterInfo
	 * @return
	 */
	private SyncCallBack receiveAdapterBaseSyncData(NodeInfo fromNodeInfo,
			NodeInfo srcNodeInfo, AdapterBase adapterBase) {
		SyncCallBack syncCallBack = new SyncCallBack();
		try {
			adapterBaseService.insertOrUpdate(adapterBase);
			
			adapterBaseService.insertByAdapterBase(adapterBase);
			
			
			syncToAllNodes(SyncType.ADAPTER_BASE,isFromParent(fromNodeInfo),fromNodeInfo,srcNodeInfo,adapterBase,SyncOperType.UPDATE,true);
			syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
		} catch (Exception e) {
			syncCallBack.setMsg("目的节点接收同步的适配器基础信息失败,"+e.getMessage());
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
		}
		return syncCallBack;
	}
	
	/**
	 * 接收适配器同步信息
	 * @param fromNode
	 * @param srcNode
	 * @param adapterInfo
	 * @return
	 */
	private SyncCallBack receiveAdapterInfoSyncData(NodeInfo fromNodeInfo,
			NodeInfo srcNodeInfo, AdapterInfo adapterInfo) {
		SyncCallBack syncCallBack = new SyncCallBack();
		try {
			adapterInfo.setIslocal(YESNO.NO.getStatus());
			adapterInfoService.insertOrUpdate(adapterInfo);
			syncToAllNodes(SyncType.ADAPTER,isFromParent(fromNodeInfo),fromNodeInfo,srcNodeInfo,adapterInfo,SyncOperType.UPDATE,true);
			syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
		} catch (Exception e) {
			syncCallBack.setMsg("目的节点接收同步的适配器绑定信息失败,"+e.getMessage());
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
		}
		return syncCallBack;
	}
	//接收网闸权限信息
	public SyncCallBack receiveGapCongInfo(GapOwnerConfigSyncVo configSyncVo){
		//先根据网闸所有者删除，然后批量添加，同步过来的网闸都是一个所有者的
		final SyncCallBack syncCallBack = new SyncCallBack();
		if(configSyncVo!=null){
			final String owner = configSyncVo.getOwnerCode();
			final List<GapOwnerConf>  listGapOwnerConf = configSyncVo.getListGapOwnerConf();
			try {
				CM.getDao().doInSingleTransationCircle("处理注册同步信息", new SingleTransationCircleWithOutResult() {
	
					@Override
					public void actionInCircle() throws RuntimeException {
						try {
							gapOwnerConfService.deleteGapOwnerConfigByOwnerCode(owner);
							if(CollectionUtil.hasElement(listGapOwnerConf)){
								gapOwnerConfService.insertBatch(listGapOwnerConf);
							}
							syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
						}catch (Exception e) {
							throw new RuntimeException(e);
						}
					}
					
				});
			} catch (Exception e) {
				//有异常给管理节点反馈注册失败
				syncCallBack.setResult(SUCCESSFAIL.FAIL);
				syncCallBack.setMsg("节点信息同步处理失败！");
			}
		}
		return syncCallBack;
	}
	/**
	 * 删除交换关系
	 * @param exchRel
	 * @return
	 */
	private SyncCallBack receiveDeleteRelSyncData(ExchRel exchRel) {
		SyncCallBack syncCallBack = new SyncCallBack();
		try {
		
			exchRelService.delete(exchRel.getNodeIdCreator(), exchRel.getNodeIdTo());
			syncCallBack.setMsg("目的节点接收删除交换关系信息成功");
			syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
		} catch (ServiceException e) {
			syncCallBack.setMsg("目的节点接收删除交换关系信息失败,"+e.getMessage());
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
		}
		return syncCallBack;
	}
	/**
	 * 接收节点注册时的同步信息
	 * @param nodeRegistSyncData
	 * @return 
	 */
	public SyncCallBack receiveNodeRegistSyncData(NodeInfo fromNodeInfo,NodeInfo srcNodeInfo,  NodeRegistSyncData nodeRegistSyncData) {
		 final SyncCallBack registResult = new SyncCallBack();
		 final NodeInfo registNode = nodeRegistSyncData.getRegistNode();
		 registNode.setHasGap(NodeInfoParam.hasGap);
		 final List<NodeInfo> allNodeInfos = nodeRegistSyncData.getAllNodeInfos();//所有的已有节点的信息
		 //final List<SysInfo> allSysInfos = nodeRegistSyncData.getAllSysInfos();//所有的已注册信息
		 final List<DomainInfo> allDomainInfos = nodeRegistSyncData.getAllDomainInfos();//所有的域信息
		 final List<AdapterInfo> allAdapterInfos = nodeRegistSyncData.getAllAdapters();//所有的适配器信息
		 final List<DataCategory> allDataCategories = nodeRegistSyncData.getAllCategories();//所有的数据分类信息
		 final List<AdapterBase> allAdapterBases = nodeRegistSyncData.getAllAdaterBases();//所有的适配器信息基本表
		 try {
			//验证参数 ：同一个行政区内不能出现两个管理节点。
			String fromNodeReg = fromNodeInfo.getNodeCode().substring(0, 6);
			String registReg = registNode.getNodeCode().substring(0, 6);
			if(fromNodeReg.trim().equalsIgnoreCase(registReg.trim())){
				if(NodeInfoParam.isManaagerNode()){
					//有异常给管理节点反馈注册失败
					registResult.setResult(SUCCESSFAIL.FAIL);
					registResult.setMsg("节点注册失败：同一个行政区代码下不能有两个管理节点，请修改目的节点为非管理节点再进行注册！");
					return registResult;
				}
			}
			System.out.println("--------------节点注册!!");
			CM.getDao().doInSingleTransationCircle("处理注册同步信息", new SingleTransationCircleWithOutResult() {
				@Override
				public void actionInCircle() throws RuntimeException {
					try {
						registNode.setIsLocal(YESNO.YES.getStatus());
						registNode.setHasGap(NodeInfoParam.hasGap);
						NodeInfo nodeInfo =  nodeInfoService.getNodeByNodeCode(registNode.getNodeCode());
						if(nodeInfo!=null){
							registNode.setId(nodeInfo.getId());
						}else{
							NodeInfo localNode  = nodeInfoService.getLocalNode();
							if(localNode != null){
								registResult.setMsg("节点访问地址已被["+localNode.getNodeName()+"]占用!");
								throw new RuntimeException("节点访问地址已被["+localNode.getNodeName()+"]占用!");
							}
						}
						nodeInfoService.insertOrUpdate(registNode);
						System.out.println("----------------节点注册startlocalConfCallBack.onLocalNodeInit()");
						localConfCallBack.onLocalNodeInit(registNode);
						System.out.println("----------------节点注册endlocalConfCallBack.onLocalNodeInit()");
						setNodesInfoToUnLocalFlag(allNodeInfos, registNode);
						insertOrUpdateSyncNodes(allNodeInfos);
						//insertOrUpdateSyncSysInfo(allSysInfos);
						//setNodesPipeLinesToUnLocalFlag(allPipeInfos, registNode);
						//insertOrUpdateSyncPipeLineInfo(allPipeInfos);
						insertOrUpdateSyncDomains(allDomainInfos);
						setAdaptersToUnLocalFlag(allAdapterInfos,registNode);
						insertOrUpdateSyncAdapters(allAdapterInfos);
						insertOrUpdateSyncAdapterBases(allAdapterBases);
						insertOrUpdateSyncDataCategories(allDataCategories);
						//注册后需要将消息队列和ftp的通道信息反馈回去，以便建立交换关系用。
						NodeRegistSyncData nodeRegistSyncData = new NodeRegistSyncData();
						registNode.setIsLocal("0");
						registResult.setResult(SUCCESSFAIL.SUCCESS);
						//nodeRegistSyncData.setAllPipeInfos(nodePipelineService.getLocalNodePipeline());
						nodeRegistSyncData.setRegistNode(registNode);
						// registResult.setCallBackObj(nodeRegistSyncData);
						 registResult.setNodeRegistSyncData(nodeRegistSyncData);
					} catch (Exception e) {
						throw new RuntimeException(e.getMessage());
					}
				}
			});
		 
		 
		} catch (Exception e) {
			//有异常给管理节点反馈注册失败
			registResult.setResult(SUCCESSFAIL.FAIL);
			registResult.setMsg("节点注册失败！"+registResult.getMsg());
			return registResult;
		}
		try {
			 //成功开始向做全网同步逻辑。
			 syncToAllNodes(SyncType.NODE_INFO,true,fromNodeInfo,srcNodeInfo,registNode,SyncOperType.INSERT,true);
		} catch (Exception e) {
			//同步失败并不影响注册，可手动同步
		}
		
		 return registResult;
	}
	/**
	 * 接收同步的节点基本信息
	 * @param nodeInfo
	 * @param srcNodeInfo 
	 * @return
	 */
	public SyncCallBack receiveNodeInfoSyncData(NodeInfo fromNodeInfo, NodeInfo srcNodeInfo) {
		SyncCallBack syncCallBack = new SyncCallBack();
		
		try {
			NodeInfo localnode = nodeInfoService.getLocalNode();
			if(localnode==null){
				throw new Exception("目的节点["+srcNodeInfo.getNodeName()+"("+srcNodeInfo.getNodeCode()+")]在地址["+srcNodeInfo.getInstallUrl()+"]不存在");
			}
			if(localnode.getId().equalsIgnoreCase(srcNodeInfo.getId())){
				srcNodeInfo.setIsLocal(YESNO.YES.getStatus());
				srcNodeInfo.setHasGap(NodeInfoParam.hasGap);//是否有网闸
			}else{
				srcNodeInfo.setIsLocal(YESNO.NO.getStatus());
			}
			nodeInfoService.insertOrUpdate(srcNodeInfo);
			syncToAllNodes(SyncType.NODE_INFO,isFromParent(fromNodeInfo),fromNodeInfo,srcNodeInfo,srcNodeInfo,SyncOperType.UPDATE,true);
			syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
			turboCacheService.updateNodeCache(CacheOperType.UPDATE, srcNodeInfo);
		} catch (Exception e) {
			syncCallBack.setMsg("目的节点接收同步的节点信息失败,"+e.getMessage());
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
		}
		return syncCallBack;
	}
	/**
	 * 接收同步的节点基本信息
	 * @param nodeInfo
	 * @return
	 */
	public SyncCallBack receiveDomainInfoSyncData(NodeInfo fromNodeInfo, NodeInfo srcNodeInfo,DomainInfo domainInfo) {
		SyncCallBack syncCallBack = new SyncCallBack();
		try {
			domainInfoService.insertOrUpdate(domainInfo);
			syncToAllNodes(SyncType.DOMAIN_INFO,isFromParent(fromNodeInfo),fromNodeInfo,srcNodeInfo,domainInfo,SyncOperType.UPDATE,true);
			syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
		} catch (Exception e) {
			syncCallBack.setMsg("目的节点接收同步的业务域信息失败,"+e.getMessage());
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
		}
		return syncCallBack;
	}
	
	/**
	 * 接收同步的节点基本信息
	 * @param nodeInfo
	 * @return
	 */
	public SyncCallBack receivePipelineSyncData(NodeInfo fromNodeInfo,NodeInfo srcNodeInfo,NodePipeline pipeline) {
		SyncCallBack syncCallBack = new SyncCallBack();
		pipeline.setIsLocal(YESNO.NO.getStatus());
		try {
			//nodePipelineService.insertOrUpdate(pipeline);
			tlqMoveService.updateRemote(pipeline);
			syncToAllNodes(SyncType.PILE_LINE,isFromParent(fromNodeInfo),fromNodeInfo,srcNodeInfo,pipeline,SyncOperType.UPDATE,true);
			syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
		} catch (Exception e) {
			syncCallBack.setMsg("目的节点接收同步的通道信息失败,"+e.getMessage());
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
		}
		return syncCallBack;
	}
	/**
	 * 接收同步的节点基本信息
	 * @param nodeInfo
	 * @return
	 */
	public SyncCallBack receivePipelineListSyncData(NodeInfo fromNodeInfo,NodeInfo srcNodeInfo,List<NodePipeline> nodePipelines) {
		SyncCallBack syncCallBack = new SyncCallBack();
		setNodesPipeLinesToUnLocalFlag(nodePipelines, null);
		try {
			for (Iterator<NodePipeline> iterator = nodePipelines.iterator(); iterator
					.hasNext();) {
				NodePipeline nodePipeline = iterator.next();
				//nodePipelineService.insertOrUpdate(nodePipeline);
				tlqMoveService.updateRemote(nodePipeline);
			}
			
			syncToAllNodes(SyncType.PILE_LINE_LIST,isFromParent(fromNodeInfo),fromNodeInfo,srcNodeInfo,nodePipelines,SyncOperType.UPDATE,true);
			syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
		} catch (Exception e) {
			syncCallBack.setMsg("目的节点接收同步的通道信息失败,"+e.getMessage());
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
		}
		return syncCallBack;
	}
	/**
	 * 处理建立交换关系
	 * @param exchRel
	 * @return
	 */
	private SyncCallBack receiveCreateRelSyncData(ExchRel exchRel) {
		SyncCallBack syncCallBack = new SyncCallBack();
		try {
			exchRelService.addRelData(exchRel);
		} catch (ServiceException e) {
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
			return syncCallBack;
		}
		syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
		try {
			NodeInfo localNode = nodeInfoService.getLocalNode();
			NodeInfo creatorNode = nodeInfoService.get(exchRel.getNodeIdCreator());
			NodeInfo toNodeInfo = nodeInfoService.get(exchRel.getNodeIdTo());
			NodeInfo remoteNode = toNodeInfo;
			if(creatorNode.getId().equalsIgnoreCase(localNode.getId())){
				remoteNode = toNodeInfo;
			}
			if(toNodeInfo.getId().equalsIgnoreCase(localNode.getId())){
				remoteNode = creatorNode;
			}
			localConfCallBack.onNodeExchRelCreate(localNode, remoteNode);
			syncCallBack.setStatus(true);
		} catch (Exception e) {
			syncCallBack.setStatus(false);
		}
		//判断是否是和本地节点有关的交换关系
		return syncCallBack;
	}
	/**
	 * 同步数据给节点
	 * @param syncType
	 * @param toNode
	 * @param dataObject
	 * @param syncOperType
	 * @throws NiceException
	 * @throws ServiceException
	 */
	public void syncToNodeFromLocalNode(SyncType syncType,NodeInfo toNode,Object dataObject,SyncOperType syncOperType,boolean whenErrorRepeat) throws NiceException {
		try {
			NodeInfo localNode = nodeInfoService.getLocalNode();
			if(localNode!=null){
			//从注册开始同步数据的发起节点变为自己
			syncCenter.registSyncTaskAndStart(UUIDGenerator.getUUID(), localNode,localNode, toNode, JsonUtil.toJson(dataObject), syncOperType, syncType,whenErrorRepeat);
			}
		} catch (ServiceException e) {
			throw new NiceException("注册同步任务失败",e);
		}
	}
	/**
	 * 全网同步
	 * @param syncType
	 * @param registNode
	 * @throws NiceException 
	 */
	public void syncToAllNodes(SyncType syncType,boolean fromParent,NodeInfo fromNode,NodeInfo srcNode,Object dataObject,SyncOperType syncOperType,boolean whenErrorRepeat) throws NiceException {
		String sn = UUIDGenerator.getUUID();
		 //1 将当前注册的节点信息同步给下级节点（非当前节点）
		 //2如果是上级节点同步下来的，当前节点收到后只往下同步。
		 //3如果是下级节点同步上来的，将当前注册的节点信息要同步给上级节点，和下级节点（排除信息源节点）。
		List<NodeInfo> toNodes = new ArrayList<NodeInfo>();
		if(fromParent){
			List<NodeInfo> childNodes = nodeRegistryPermissionMgr.getManagedNodes();
			if(CollectionUtil.hasElement(childNodes)){
				toNodes.addAll(childNodes);
			}
		}else{
			NodeInfo parentNode = nodeRegistryPermissionMgr.getManagerNode();
			if(parentNode!=null){
				toNodes.add(parentNode);
			}
			List<NodeInfo> childNodes = nodeRegistryPermissionMgr.getManagedNodes();
			if(CollectionUtil.hasElement(childNodes)){
				for (Iterator<NodeInfo> iterator = childNodes.iterator(); iterator.hasNext();) {
					NodeInfo nodeInfo =  iterator.next();
					if(nodeInfo.getId().equalsIgnoreCase(fromNode.getId())){
						iterator.remove();
						break;
					}
				}
				toNodes.addAll(childNodes);
			}
		}
		if(CollectionUtil.hasElement(toNodes)){
			//----Add by Macky_He on 2016/01/04
			//同步时过滤掉相同的url地址的信息
			String syncUrl = "";
			for (int i = 0; i < toNodes.size(); i++){
				syncUrl = toNodes.get(i).getInstallUrl();
				for (int j = toNodes.size() - 1; j > i; j--){
					if (toNodes.get(j).getInstallUrl().equalsIgnoreCase(syncUrl)){
						toNodes.remove(j);
					}
				}
			}
			NodeInfo localNode;
			try {
				//从注册开始同步数据的发起节点变为自己
				localNode = nodeInfoService.getLocalNode();
				if (whenErrorRepeat) {
					syncCenter.registSyncTasksAndStart(sn, localNode,srcNode, toNodes, JsonUtil.toJson(dataObject), syncOperType, syncType,whenErrorRepeat);
				}else{}
			} catch (ServiceException e) {
				throw new NiceException("注册同步任务失败",e);
			}
			
		}
	}
	/**
	 * 全网同步
	 * @param syncType
	 * @param registNode
	 * @throws ServiceException 
	 * @throws NiceException 
	 */
	public void syncToAllNodesFromLocalNode(SyncType syncType,boolean fromParent,Object dataObject,SyncOperType syncOperType,boolean whenErrorRepeat) throws ServiceException, NiceException   {
	
		NodeInfo fromNode = nodeInfoService.getLocalNode();
		if(fromNode!=null){
			syncToAllNodes(syncType, fromParent, fromNode,fromNode, dataObject, syncOperType,whenErrorRepeat);
		}
	}
	//-----------------------------------------tools
	protected void insertOrUpdateSyncDataCategories(
			List<DataCategory> allDataCategories) throws ServiceException {
		try {
			if (CollectionUtil.hasElement(allDataCategories)) {
				for (Iterator<DataCategory> iterator = allDataCategories.iterator(); iterator.hasNext();) {
					DataCategory dataCategory = iterator.next();
					insertOrUpdateRemoteataCategorie(dataCategory);
				}
			}
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
		
	}
	private void insertOrUpdateRemoteataCategorie(DataCategory dataCategory) throws ServiceException {
		try {
			categoryService.insertOrUpdateRemoteNode(dataCategory);
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
		
	}
	
	
	protected void insertOrUpdateSyncAdapterBases(List<AdapterBase> allAdapterBases) throws ServiceException {
		try {
			if (CollectionUtil.hasElement(allAdapterBases)) {
				for (Iterator<AdapterBase> iterator = allAdapterBases.iterator(); iterator.hasNext();) {
					AdapterBase adapterBase = iterator.next();
					insertOrUpdateRemoteAdapterBase(adapterBase);
				}
			}
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
		
	}
	
	
	protected void insertOrUpdateSyncAdapters(List<AdapterInfo> allAdapterInfos) throws ServiceException {
		try {
			if (CollectionUtil.hasElement(allAdapterInfos)) {
				for (Iterator<AdapterInfo> iterator = allAdapterInfos.iterator(); iterator.hasNext();) {
					AdapterInfo adapterInfo = iterator.next();
					insertOrUpdateRemoteAdapterInfo(adapterInfo);
				}
			}
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
		
	}
	
	private void insertOrUpdateRemoteAdapterBase(AdapterBase adapterBase) throws ServiceException {
		try {
			adapterBaseService.insertOrUpdateRemoteNode(adapterBase);
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
		
	}
	
	private void insertOrUpdateRemoteAdapterInfo(AdapterInfo adapterInfo) throws ServiceException {
		try {
			adapterInfoService.insertOrUpdateRemoteNode(adapterInfo);
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
		
	}
	public void insertOrUpdateSyncNode(NodeInfo remoteNode) throws ServiceException {
		try {
			nodeInfoService.insertOrUpdateRemoteNode(remoteNode);
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
	}
	public void insertOrUpdateSyncNodes(List<NodeInfo> syncNodes) throws ServiceException {
		try {
			if (CollectionUtil.hasElement(syncNodes)) {
				for (Iterator<NodeInfo> iterator = syncNodes.iterator(); iterator
						.hasNext();) {
					NodeInfo nodeInfo =  iterator.next();
					insertOrUpdateSyncNode(nodeInfo);
				}
			}
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
	}
	public void insertOrUpdateRemotePipeLineInfo(NodePipeline pipeLine) throws ServiceException {
		try {
			nodePipelineService.insertOrUpdateRemotePipeLine(pipeLine);
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
	}
	public void insertOrUpdateRemotePipeLineInfo(List<NodePipeline> pipeLines) throws ServiceException {
		try {
			if (CollectionUtil.hasElement(pipeLines)) {
				for (Iterator<NodePipeline> iterator = pipeLines.iterator(); iterator.hasNext();) {
					NodePipeline pipeLine = iterator.next();
					insertOrUpdateRemotePipeLineInfo(pipeLine);
				}
			}
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
	}
	/*public void insertOrUpdateSyncSysInfo(SysInfo sysInfo) throws ServiceException {
		try {
			sysInfoService.saveOrUpdateRemoteSysInfo(sysInfo);
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
	}

	public void insertOrUpdateSyncSysInfo(List<SysInfo> sysInfos) throws ServiceException {
		try {
			if (CollectionUtil.hasElement(sysInfos)) {
				for (Iterator<SysInfo> iterator = sysInfos.iterator(); iterator.hasNext();) {
					SysInfo sysInfo = iterator.next();
					insertOrUpdateSyncSysInfo(sysInfo);
				}
			}
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}

	}*/
	public void insertOrUpdateSyncPipeLineInfo(NodePipeline pipeLine) throws ServiceException {
		try {
			nodePipelineService.insertOrUpdateRemotePipeLine(pipeLine);
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
	}

	public void insertOrUpdateSyncPipeLineInfo(List<NodePipeline> pipeLines) throws ServiceException {
		try {
			if (CollectionUtil.hasElement(pipeLines)) {
				for (Iterator<NodePipeline> iterator = pipeLines.iterator(); iterator.hasNext();) {
					NodePipeline pipeLine = iterator.next();
					insertOrUpdateRemotePipeLineInfo(pipeLine);
				}
			}
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
	}
	public void insertOrUpdateSyncDomains(List<DomainInfo> domainInfos) throws ServiceException {
		try {
			if (CollectionUtil.hasElement(domainInfos)) {
				for (Iterator<DomainInfo> iterator = domainInfos.iterator(); iterator.hasNext();) {
					DomainInfo domainInfo = iterator.next();
					insertOrUpdateSyncDomainNode(domainInfo);
				}
			}
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
	}
	public void insertOrUpdateSyncDomainNode(DomainInfo domainInfo) throws ServiceException {
		try {
			domainInfoService.insertOrUpdate(domainInfo);
		} catch (ServiceException e) {
			throw new ServiceException(e);
		}
	}
	public boolean isFromParent(NodeInfo fromNodeInfo) throws ServiceException{
		NodeInfo localNodeInfo = nodeInfoService.getLocalNode();
		if(localNodeInfo.getPid().equalsIgnoreCase(fromNodeInfo.getId())){
			return true;
		}else{
			return false;
		}
	}
	public void insertOrUpdateSynchronizeInfo(SynchronizeInfo synchronizeInfo) throws ServiceException {
		try {
			if(StringUtils.hasText(synchronizeInfo.getId())){
				CM.getDao().saveOrUpdateModuleById("修改同步信息", synchronizeInfo, null);
			}
		} catch (DaoException e) {
			throw new ServiceException(e);
		}
	}
	public void setNodesInfoToUnLocalFlag(List<NodeInfo> allNodeInfos,NodeInfo toLocalNode){
		if(CollectionUtil.hasElement(allNodeInfos)){
			for (Iterator<NodeInfo> iterator = allNodeInfos.iterator(); iterator
					.hasNext();) {
				NodeInfo nodeInfo =  iterator.next();
				if(toLocalNode!=null&&nodeInfo.getId().equalsIgnoreCase(toLocalNode.getId())){
					nodeInfo.setIsLocal(YESNO.YES.getStatus());
				}else{
					nodeInfo.setIsLocal(YESNO.NO.getStatus());
				}
			}
		}
	}
	public void setNodesPipeLinesToUnLocalFlag(List<NodePipeline> nodePipelines,NodeInfo toLocalNode){
		if(CollectionUtil.hasElement(nodePipelines)){
			for (Iterator<NodePipeline> iterator = nodePipelines.iterator(); iterator
					.hasNext();) {
				NodePipeline nodePipeline =  iterator.next();
				if(toLocalNode!=null&&nodePipeline.getNodeId().equalsIgnoreCase(toLocalNode.getId())){
					nodePipeline.setIsLocal(YESNO.YES.getStatus());
				}else{
					nodePipeline.setIsLocal(YESNO.NO.getStatus());
				}
			}
		}
	}
	public void setAdaptersToUnLocalFlag(List<AdapterInfo> adapters,NodeInfo toLocalNode){
		if(CollectionUtil.hasElement(adapters)){
			for (Iterator<AdapterInfo> iterator = adapters.iterator(); iterator
					.hasNext();) {
				AdapterInfo adapter =  iterator.next();
				if(toLocalNode!=null&&adapter.getNodeCode().equalsIgnoreCase(toLocalNode.getNodeCode())){
					adapter.setIslocal(YESNO.YES.getStatus());
				}else{
					adapter.setIslocal(YESNO.NO.getStatus());
				}
			}
		}
	}
	static Object saveReceiveMonitorSyncData_lock = new Object();
	public SyncCallBack receiveMonitorSyncData(NodeInfo toNode,QuotaValueOnTime quotaValueOnTime) {
		SyncCallBack syncCallBack = new SyncCallBack();
		try {
			//@TODO 同一个id对象传输太快，会照成并发问题
			synchronized (saveReceiveMonitorSyncData_lock) {
//				System.out.println(toNode.getNodeCode()+"-----from---------quotaValueOnTime::::"+quotaValueOnTime);
				monitorService.insertOrUpdateQuotaOnTime(quotaValueOnTime);
			}
//			syncToNodeFromLocalNode(SyncType.MONITOR, toNode, quotaValueOnTime, SyncOperType.UPDATE);
			syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
		} catch (Exception e) {
			syncCallBack.setMsg("目的节点接收同步的监控信息失败,"+e.getMessage());
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
		}
		return syncCallBack;
	}
	
	
	/**
	 * 通过节点，获取交换节点的tlq信息
	 * @return
	 * @throws ServiceException 
	 */
	public NodePipeline getPipleLineInfo(String destNode) throws ServiceException{
		String msg="";
		try {
			NodePipeline nodePipeline=pipelineRegionService.getMinLoad(destNode);
			return nodePipeline;
		} catch (ServiceException e) {
			msg="建立交换关系时获取失败";
			log.error(msg, e);
			throw new ServiceException(msg);
		}
	}
	
	/**
	 * 通过节点，获取交换节点的tlq信息
	 * @return
	 * @throws ServiceException 
	 */
	public void updatePipeline(NodePipeline nodePipeline) throws ServiceException{
		nodePipeline.setIsLocal("0");
	    nodePipelineService.insertOrUpdate(nodePipeline);//更新对方的通道信息
	}
	
}
